package com.daring.mysql2maxkb;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONException;
import com.alibaba.fastjson.JSONObject;
import com.mashape.unirest.http.HttpResponse;
import com.mashape.unirest.http.Unirest;
import com.mashape.unirest.http.exceptions.UnirestException;
import okhttp3.*;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.io.IOException;
import java.sql.*;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

public class MySQL2MaxKBParagraph {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 增量读取 MySQL 数据
        DataStream<JSONObject> dataStream = env
                .addSource(new RichSourceFunction<JSONObject>() {
                    private static final long serialVersionUID = 3334654984018091675L;

                    private Connection connect = null;
                    private PreparedStatement ps = null;
                    private PreparedStatement updps = null;
                    private long index = 0L;

                    private boolean running=false;
                    /*
                     * (non-Javadoc)
                     *
                     * @see org.apache.flink.api.common.functions.AbstractRichFunction#open(org.
                     * apache.flink.configuration.Configuration) to use open database connect
                     */
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        running =true;
                        Class.forName("com.mysql.jdbc.Driver");
                        connect = (Connection) DriverManager.getConnection("jdbc:mysql://192.168.0.213:3311/nngsl?useUnicode=true&characterEncoding=UTF-8&useAffectedRows=true&allowMultiQueries=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8", "root", "123456");
                        ps = (PreparedStatement) connect.prepareStatement("SELECT * FROM cms_content WHERE id > ?");
                    }

                    @Override
                    public void run(SourceContext<JSONObject> sourceContext) throws Exception {
                        while(running){
                            ps.setLong(1, index);
                            ResultSet resultSet = ps.executeQuery();
                            while (resultSet.next()) {
                                JSONObject jsonObject = resultSetToJsonObject(resultSet);
                                long id = resultSet.getLong("id");
                                sourceContext.collect(jsonObject);
                                index = id;
                            }
                        }
                    }

                    public JSONObject resultSetToJsonObject(ResultSet rs) throws SQLException, JSONException {
                        // json对象
                        JSONObject jsonObj = new JSONObject();
                        // 获取列数
                        ResultSetMetaData metaData = rs.getMetaData();
                        int columnCount = metaData.getColumnCount();
                        // 遍历ResultSet中的每条数据
                        // 遍历每一列
                        for (int i = 1; i <= columnCount; i++) {
                            String columnName =metaData.getColumnLabel(i);
                            String value = rs.getString(columnName);
                            jsonObj.put(columnName, value);
                        }
                        return jsonObj;

                    }

                    /*
                     * (non-Javadoc)
                     *
                     * @see
                     * org.apache.flink.streaming.api.functions.source.SourceFunction#cancel()
                     * colse database connect
                     */
                    @Override
                    public void cancel() {
                        running =false;
                        try {
                            super.close();
                            if (connect != null) {
                                connect.close();
                            }
                            if (ps != null) {
                                ps.close();
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                    }
                }).name("增量读取 MySQL 数据").setParallelism(1);

        // 摘出分段内容
        SingleOutputStreamOperator<String> extractSegmentedMainTextStream = dataStream.map(new MapFunction<JSONObject, String>() {
            @Override
            public String map(JSONObject jsonObject) throws Exception {
                //return jsonObject.getString("content_details");
                return HtmlToTextConverter.convertHtmlToText(jsonObject.getString("content_details"));

            }
        }).name("摘取分段主文本").setParallelism(1).filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                return s != null && !s.trim().equals("") && !s.replace("    ", "").replace(" ", "").replace("cc", "").equals("");
            }
        }).name("过滤空值").setParallelism(1);

        // ai总结标题
        SingleOutputStreamOperator<HashMap<String, String>>  aiTitleStream = extractSegmentedMainTextStream.map(new MapFunction<String, HashMap<String, String>>() {
            private static final long serialVersionUID = 3334654984018321675L;
            @Override
            public HashMap<String, String> map(String s)  throws IOException, UnirestException{
               return AIUtils.generateTitle2(s);
            }
        }).name("ai总结标题").setParallelism(1);


        /*extractSegmentedMainTextStream.addSink(new SinkFunction<String>() {
            @Override
            public void invoke(String value, Context context) throws Exception {
                SinkFunction.super.invoke(value, context);
                System.out.println("value = " + value);
            }
        }).name("临时打印").setParallelism(1);*/

        // ai生成问题
        SingleOutputStreamOperator<HashMap<String, Object>> aiQuestStream = aiTitleStream.map(new MapFunction<HashMap<String, String>, HashMap<String, Object>>() {
            @Override
            public HashMap<String, Object> map(HashMap<String, String> stringStringHashMap) throws Exception {
                return AIUtils.generateQuests(stringStringHashMap);
            }
        }).name("ai生成问题").setParallelism(1);

        // 构建文档追加请求体


        // 写出到 MaxKB 中
        aiQuestStream.addSink(new RichSinkFunction<HashMap<String, Object>>() {
            @Override
            public void invoke(HashMap<String, Object> value, Context context) throws Exception {
                super.invoke(value, context);
                System.out.println("value = " + value);
                MaxKBUtils.request2AddSegmentation(value);



            }
        }).setParallelism(10);
        env.execute();

    }
}
